package ex.ex2;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;

/**
 * 关键字是纯数字，电话舆情
 */
public class NumberBoltB extends BaseBasicBolt {
    private final List<String> keywords = new ArrayList<String>(){
        {
            add("12315");
            add("12345");
        }
    };

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        String sentence = input.getStringByField("sentence");
        for (String keyword : keywords) {
            if (sentence.contains(keyword)) {
                collector.emit(new Values(input.getValue(0), sentence + " 电话舆情"));
                return;
            }
        }
        collector.emit(new Values(input.getValue(0), sentence + " "));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("id", "sentence"));
    }
}
